package cn.lsh.spark.sql;

import cn.lsh.conts.UserAction;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.Serializable;
import java.util.*;

/**
 * 通过Spark算子实现协同过滤算法的物品推荐
 */
public class ItemUserSpark {

	public static void main(String[] args) {
		SparkSession ss = SparkSession.builder().master("local")
				.appName("item_user")
				.getOrCreate();
		JavaSparkContext jsc = new JavaSparkContext(ss.sparkContext());
		JavaRDD<String> textRdd = jsc.textFile("file:/bigdata/hadoop-test/input/itemCF/test/input/test_data.txt");
		/*@step1 加载测试数据转换为二元组rdd：<itemId,<userId,record>>*/
		JavaPairRDD<String, Tuple2<String, Integer>> userItemRecordRdd = textRdd.distinct().mapToPair(line -> new Tuple2<>(line.split(",")[0],
				new Tuple2<>(line.split(",")[1], UserAction.getRecord(line.split(",")[2]))));
		/*@step2 计算各个用户对物品的评分：<itemId,Map<userId,record>>*/
		JavaPairRDD<String, HashMap<String, Integer>> userItemRecordCountRdd = userItemRecordRdd.aggregateByKey(new HashMap<>(), new Function2<HashMap<String, Integer>, Tuple2<String, Integer>, HashMap<String, Integer>>() {
			private static final long serialVersionUID = 4896422919000738408L;

			@Override
			public HashMap<String, Integer> call(HashMap<String, Integer> v1, Tuple2<String, Integer> v2) throws Exception {
				v1.merge(v2._1, v2._2, Integer::sum);
				return v1;
			}
		}, new Function2<HashMap<String, Integer>, HashMap<String, Integer>, HashMap<String
				, Integer>>() {
			private static final long serialVersionUID = 2135196741108911099L;

			@Override
			public HashMap<String, Integer> call(HashMap<String, Integer> v1, HashMap<String, Integer> v2) throws Exception {
				for (Map.Entry<String, Integer> entry : v2.entrySet()) {
					v1.merge(entry.getKey(), entry.getValue(), Integer::sum);
				}
				return v1;
			}
		});
		userItemRecordCountRdd.cache();
		/*@step3 转换为统计后的用户对物品评分rdd：<itemId,<userId,record>>*/
		JavaPairRDD<String, Tuple2<String, Integer>> itemUserRecordRdd = userItemRecordCountRdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, HashMap<String, Integer>>>, String, Tuple2<String, Integer>>() {
			private static final long serialVersionUID = -2318624350497226532L;

			@Override
			public Iterator<Tuple2<String, Tuple2<String, Integer>>> call(Iterator<Tuple2<String, HashMap<String, Integer>>> tuple2Iterator) throws Exception {
				List<Tuple2<String, Tuple2<String, Integer>>> ret = new ArrayList<>();
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, HashMap<String, Integer>> next = tuple2Iterator.next();
					for (Map.Entry<String, Integer> entry : next._2.entrySet()) {
						ret.add(new Tuple2<>(entry.getKey(), new Tuple2<>(next._1, entry.getValue())));
					}
				}
				return ret.iterator();
			}
		});
		/*@step4 生成物品对物品同现次数的二元组rdd：<<itemId,itemId>, 1>*/
		JavaPairRDD<Tuple2<String, String>, Integer> itemItemTmpRdd = userItemRecordCountRdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, HashMap<String, Integer>>>, Tuple2<String, String>, Integer>() {
			private static final long serialVersionUID = -2318624350497226532L;

			@Override
			public Iterator<Tuple2<Tuple2<String, String>, Integer>> call(Iterator<Tuple2<String, HashMap<String, Integer>>> tuple2Iterator) throws Exception {
				List<Tuple2<Tuple2<String, String>, Integer>> ret = new ArrayList<>();
				while (tuple2Iterator.hasNext()) {
					Set<String> itemSet = tuple2Iterator.next()._2.keySet();
					Object[] os = itemSet.toArray();
					for (int i = 0; i < os.length; i++) {
						ret.add(new Tuple2<>(new Tuple2<>(os[i].toString(), os[i].toString()), 1));
						for (int j = i + 1; j < os.length; j++) {
							ret.add(new Tuple2<>(new Tuple2<>(os[i].toString(), os[j].toString()), 1));
						}
					}
				}
				return ret.iterator();
			}
		});
		/*@step5 统计得到物品对物品同现矩阵rdd：<itemId,<itemId, count>>*/
		JavaPairRDD<String, Tuple2<String, Integer>> itemItemRdd = itemItemTmpRdd.reduceByKey(Integer::sum)
				.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<Tuple2<String, String>, Integer>>, String, Tuple2<String, Integer>>() {
					private static final long serialVersionUID = 8193629530163757444L;

					@Override
					public Iterator<Tuple2<String, Tuple2<String, Integer>>> call(Iterator<Tuple2<Tuple2<String, String>, Integer>> tuple2Iterator) throws Exception {
						List<Tuple2<String, Tuple2<String, Integer>>> ret = new ArrayList<>();
						while (tuple2Iterator.hasNext()) {
							Tuple2<Tuple2<String, String>, Integer> mapTuple2 = tuple2Iterator.next();
							ret.add(new Tuple2<>(mapTuple2._1._1, new Tuple2<>(mapTuple2._1._2, mapTuple2._2)));
							if (!mapTuple2._1._1.equals(mapTuple2._1._2)) {
								//如果两个物品不相等，需要再输出一条镜像记录
								ret.add(new Tuple2<>(mapTuple2._1._2, new Tuple2<>(mapTuple2._1._1, mapTuple2._2)));
							}
						}
						return ret.iterator();
					}
				});
		itemItemRdd.cache();
		System.out.println("**********物品同现矩阵");
		itemItemRdd.foreach(t -> System.out.println(t));
		/*@step7 用户对物品评分矩阵乘以物品同现矩阵：<itemId, <userId, record>, <itemId, count>>*/
		JavaPairRDD<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>> join = itemUserRecordRdd.join(itemItemRdd);
		/*@step8 得到矩阵行列乘积rdd：<<userId, itemId>, value>*/
		JavaPairRDD<Tuple2<String, String>, Integer> userItemMultiRdd = join.mapToPair(new PairFunction<Tuple2<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>>, Tuple2<String, String>, Integer>() {
			private static final long serialVersionUID = -2951379396437189294L;

			@Override
			public Tuple2<Tuple2<String, String>, Integer> call(Tuple2<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>> t) throws Exception {
				return new Tuple2<>(new Tuple2<>(t._2._1._1, t._2._2._1), t._2._1._2 * t._2._2._2);
			}
		});
		/*@step9 求和得到用户对每个物品的物品的推荐评分矩阵：<<userId, itemId>, value>*/
		JavaPairRDD<Tuple2<String, String>, Integer> userItemMultiRecordRdd = userItemMultiRdd.reduceByKey(Integer::sum);
		System.out.println("**********用户对物品推荐矩阵");
		userItemMultiRecordRdd.foreach(t -> System.out.println(t));
		/*@step10 转换成二元组rdd：<<userId, value>, itemId>*/
		JavaPairRDD<Tuple2<String, Integer>, String> userRecordItemRdd = userItemMultiRecordRdd.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Integer>, Tuple2<String, Integer>, String>() {
			private static final long serialVersionUID = 3046037895848145693L;

			@Override
			public Tuple2<Tuple2<String, Integer>, String> call(Tuple2<Tuple2<String, String>, Integer> t) throws Exception {
				return new Tuple2<>(new Tuple2<>(t._1._1, t._2), t._1._2);
			}
		});
		//排序聚合后的推荐列表rdd：<userId, string>
		JavaPairRDD<String, String> userItemsRdd = userRecordItemRdd.sortByKey(new UserRecordCompare())
				/*@step11 自定义比较排序器，先按照userId自然排序，然后按照评分倒序排序*/
				.mapToPair(new PairFunction<Tuple2<Tuple2<String, Integer>, String>, String, Tuple2<String, Integer>>() {
					private static final long serialVersionUID = 2210128857753846055L;

					@Override
					public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, Integer>, String> t) throws Exception {
						return new Tuple2<>(t._1._1, new Tuple2<>(t._2, t._1._2));
					}
					/*@step12 聚合排序后的每个用户的推荐列表*/
				}).aggregateByKey("", new Function2<String, Tuple2<String, Integer>, String>() {
					private static final long serialVersionUID = -2801745541613060736L;

					@Override
					public String call(String v1, Tuple2<String, Integer> v2) throws Exception {
						return v1 + "," + v2._1 + ":" + v2._2;
					}
				}, new Function2<String, String, String>() {
					private static final long serialVersionUID = -81211973327382009L;

					@Override
					public String call(String v1, String v2) throws Exception {
						return v1 + v2;
					}
				});
		System.out.println("*****用户物品推荐列表");
		userItemsRdd.foreach(t -> System.out.println(t));
	}

	public static class UserRecordCompare implements Serializable, Comparator<Tuple2<String, Integer>> {

		private static final long serialVersionUID = 1254251593402832337L;

		@Override
		public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
			int i = o1._1.compareTo(o2._1);
			if (i == 0) {
				i = o2._2.compareTo(o1._2);
			}
			return i;
		}
	}

}
